package com.example.demo.concurrent.MyThreadPool2;

import lombok.extern.slf4j.Slf4j;

import java.util.ArrayDeque;
import java.util.Deque;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

@Slf4j(topic = "c.MyBlockingQueue")
public class MyBlockingQueue<T> {
    //队列
    private Deque<T> queue = new ArrayDeque<>();

    //锁
    private ReentrantLock lock = new ReentrantLock();

    //容量
    private int capacity;

    //条件变量
    Condition emptyWaits = lock.newCondition();
    Condition fullWaits = lock.newCondition();

    public MyBlockingQueue(int capacity) {
        this.capacity = capacity;
    }

    //阻塞添加
    public void put(T task) {
        lock.lock();
        try{
            //队列满的时候,阻塞等待
            while(queue.size() == this.capacity) {
                try {
                    log.debug("等待加入任务队列{}",task);
                    fullWaits.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            log.debug("加入任务队列{}",task);
            queue.addLast(task);
            //唤醒空的休息室,可以获取了
            emptyWaits.signal();
        }finally {
            lock.unlock();
        }
    }

    //带有超时时间的,添加
    public boolean offer(T task,long timeout,TimeUnit unit) {
        lock.lock();
        try{
            //换算单位-->将毫秒改成纳秒
            long nanos = unit.toNanos(timeout);
            while(queue.size() == capacity) {
                try {
                    log.debug("等待加入任务队列{}",task);
                    if(nanos <= 0) {//超时了
                        return false;
                    }
                    nanos = fullWaits.awaitNanos(nanos);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            log.debug("加入任务队列{}",task);
            queue.addLast(task);
            //唤醒空的休息室,可以获取了
            emptyWaits.signal();
            return true;
        }finally {
            lock.unlock();
        }
    }

    //阻塞获取
    public T take() {
        lock.lock();
        try{
            while(queue.isEmpty()) {
                try {
                    emptyWaits.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            T t = queue.pollFirst();
            fullWaits.signal();
            return t;
        }finally {
            lock.unlock();
        }
    }

    //poll方法无需一直等待,超过超时时间则放弃
    public T pool(long timeout, TimeUnit unit) {
        lock.lock();
        try{
            //转换为纳秒
            long nanos = unit.toNanos(timeout);
            while(queue.isEmpty()) {
                try {
                    if(nanos <= 0) {
                        return null;
                    }
                    nanos = emptyWaits.awaitNanos(nanos);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            T t = queue.pollFirst();
            fullWaits.signal();
            return t;
        }finally {
            lock.unlock();
        }
    }

    public void tryPut(RejectPolicy<T> rejectPolicy, T task) {
        lock.lock();
        try {
            if(queue.size() == this.capacity) {
                //当阻塞队列放满了任务就执行拒绝策略,让用户决定下一步该怎么做
                rejectPolicy.reject(this,task);
            }else {
                log.debug("加入任务队列{}", task);
                queue.addLast(task);
                //唤醒空的休息室,可以获取了
                emptyWaits.signal();
            }
        }finally {
            lock.unlock();
        }
    }
}